package com.shujia.stream

import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Durations, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

object Demo3Window {
  def main(args: Array[String]): Unit = {

    //1、创建spark 环境
    var conf = new SparkConf()
      .setMaster("local[2]")
      .setAppName("wc")

    val sc = new SparkContext(conf)


    //2、创建SparkStreaming的上下文对象
    /**
      * 需要指定batch的间隔时间，多久计算一次的意思
      * 时间越长延迟就越长
      * 时间越吞吐量会降低
      *
      */
    val ssc = new StreamingContext(sc, Durations.seconds(5))

    /**
      * 指定checkpoint ，保持计算的状态，
      * 可以是hdfs的路径
      *
      */

    ssc.checkpoint("spark/data/checkpoint")

    //3、读取数据
    //构建Dstream,  被动接受数据
    val ds: ReceiverInputDStream[String] = ssc.socketTextStream("192.168.129.101", 8888)


    //处理数据
    val kvDS: DStream[(String, Int)] = ds.flatMap(_.split(",")).map((_, 1))


    /**
      *
      * 统计最近15秒单词的数量，每隔5秒统计一次
      *
      * 滑动窗口
      * 滚动窗口
      * 滑动时间和窗口大小必须是batch间隔的整数倍
      */


    /* val countDS: DStream[(String, Int)] = kvDS.reduceByKeyAndWindow(
       (x: Int, y: Int) => x + y, //聚合函数
       Durations.seconds(15), //窗口大小
       Durations.seconds(5) //滑动时间
     )*/

    /**
      * 优化窗口计算
      *
      */

    val countDS: DStream[(String, Int)] = kvDS.reduceByKeyAndWindow(
      (x: Int, y: Int) => x + y, //聚合函数
      (x: Int, y: Int) => x - y, //减去上一个窗口多余的部分
      Durations.seconds(15), //窗口大小
      Durations.seconds(5) //滑动时间
    )


    countDS.print()



    //启动任务
    ssc.start()
    //等待关闭
    ssc.awaitTermination()

  }
}
